package ex.datastream.windows;
import ex.datastream.ApiFrame;
import ex.datastream.functions.function.MyAggregateFunction;
import ex.datastream.functions.sourceFunction.SourceFuncByEvent;
import ex.vo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
/**
 * ref:https://blog.csdn.net/weixin_44760145/article/details/129252804
 * 一、onElement:每条元素都进入此方法执行
 * 1.当countState.value%20==0时，ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime() + 1L)触发onProcessingTime方法
 * 2.当countState.value()==50时，ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime() + 1L)触发onProcessingTime方法
 * 二、onProcessingTime
 * 1.当countState.value%20==0时，FIRE 触发 window
 * 2.当countState.value()==50时，清除状态，FIRE_AND_PURGE 触发窗口，然后销毁窗口
 * 三、process
 * 1、onProcessingTime执行完成之后执行process
 */
public class TriggerDemo {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .addSource(new SourceFuncByEvent())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Event>() {
                                            @Override
                                            public long extractTimestamp(Event element, long recordTimestamp) {
                                                return element.timestamp;
                                            }
                                        }
                                )
                )
                .keyBy(event -> true)
                .countWindow(60)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .trigger(
                        new Trigger<Event, GlobalWindow>() {
                            @Override
                            public TriggerResult onElement(Event element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
                                ValueState<Long> countState = ctx.getPartitionedState(new ValueStateDescriptor<Long>("countState", Types.LONG));
                                if (countState.value() == null){
                                    // 第一条元素
                                    countState.update(1L);
                                }else {
                                    countState.update(countState.value() + 1L);
                                }
                                System.out.printf("state:%s,",countState.value());
                                // 每 20 条数据触发一次窗口
                                if (countState.value() % 20 == 0){
                                    System.out.println("\n【1】满足条件：countState.value() % 20 == 0，请触发:onProcessingTime");
                                    ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime() + 1L);
                                }
                                // 窗口结束触发触发并情清空窗口中的元素
                                if (countState.value().longValue() == 50L){
                                    System.out.printf("\n【满足条件：countState.value().longValue() == 50L，窗口结束触发，并情清空窗口中的元素，");
                                    ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime()+ 1L);
                                }
                                return TriggerResult.CONTINUE;
                            }

                            @Override
                            public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {

                                ValueState<Long> countState = ctx.getPartitionedState(new ValueStateDescriptor<Long>("countState",Types.LONG));

                                if(countState.value() % 20 ==0){
                                    System.out.printf("【2】触发:onProcessingTime \n",countState.value());
                                    return TriggerResult.FIRE;
                                }else {
                                    System.out.printf("大窗口触发,当前条数：%s，数据清除，继续执行】\n",countState.value());
                                    countState.clear();
                                    return TriggerResult.FIRE_AND_PURGE;
                                }
                            }

                            @Override
                            public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                                System.out.println("onEventTime.....");
                                return TriggerResult.CONTINUE;
                            }

                            @Override
                            public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
                                ValueState<Long> countState = ctx.getPartitionedState(new ValueStateDescriptor<Long>("countState",Types.LONG));
                                if (countState.value().longValue() == 50L){
                                    System.out.printf("\n【2满足条件：countState.value().longValue() == 50L，窗口结束触发，并情清空窗口中的元素，");
                                    ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime()+ 1L);
                                }
                                System.out.println("clear=============================");
                            }
                        }
                )
                .process(
                        new ProcessWindowFunction<Event, String, Boolean, GlobalWindow>() {
                            @Override
                            public void process(Boolean aBoolean, Context context, Iterable<Event> elements, Collector<String> out) throws Exception {
                                out.collect("【3】process执行，窗口中有 " + elements.spliterator().getExactSizeIfKnown() + " 条元素");
                                System.out.println("============================================================");
                            }


                        }
                )
                .print();


        env.execute();
//        state:1,state:2,state:3,state:4,state:5,state:6,state:7,state:8,state:9,state:10,state:11,state:12,state:13,state:14,state:15,state:16,state:17,state:18,state:19,state:20,
//【1】满足条件：countState.value() % 20 == 0，请触发:onProcessingTime
//【2】触发:onProcessingTime
//【3】窗口中有 20 条元素
//                ============================================================
//        state:21,state:22,state:23,state:24,state:25,state:26,state:27,state:28,state:29,state:30,state:31,state:32,state:33,state:34,state:35,state:36,state:37,state:38,state:39,state:40,
//【1】满足条件：countState.value() % 20 == 0，请触发:onProcessingTime
//【2】触发:onProcessingTime
//【3】窗口中有 40 条元素
//                ============================================================
//        state:41,state:42,state:43,state:44,state:45,state:46,state:47,state:48,state:49,state:50,
//【满足条件：countState.value().longValue() == 50L，窗口结束触发，并情清空窗口中的元素，大窗口触发,当前条数：50，数据清除，继续执行】
//【3】窗口中有 50 条元素
//                ============================================================
//        state:1,state:2,state:3,state:4,state:5,state:6,state:7,state:8,state:9,state:10,state:11,state:12,state:13,state:14,state:15,state:16,state:17,state:18,state:19,state:20,
//【1】满足条件：countState.value() % 20 == 0，请触发:onProcessingTime
//【2】触发:onProcessingTime
//【3】窗口中有 20 条元素

    }

}

